package tcp

import (
	"errors"
	"fmt"
	"net"
	"os"
	"smart-flow/container/config"
	"smart-flow/container/logger"
	"sync/atomic"
	"time"

	"go.uber.org/zap"
)

type ServerManager struct {
	servers        []*EchoTcpServer
	byteTotalCount int64
}

func (m *ServerManager) Start() bool {
	for i := 0; i < config.TcpPortCount; i++ {
		m.startOne(config.TcpPortStartPos + i)
	}
	if len(m.servers) < config.TcpPortCount/2 {
		m.Stop()
		return false
	}
	go m.Stat()
	return true
}

func (m *ServerManager) startOne(port int) {
	server := new(EchoTcpServer)
	if !server.Start(port) {
		logger.Logger.Error("[容器] tcp 服务启动错误", zap.String("ip", config.IpAddr), zap.Int("port", port))
		return
	}
	m.servers = append(m.servers, server)
	config.AddTcpPort(port)
	logger.Logger.Info("[容器] tcp 服务启动成功", zap.String("ip", config.IpAddr), zap.Int("port", port))
}

func (m *ServerManager) Stop() {
	for _, server := range m.servers {
		server.Stop()
	}
}

func (m *ServerManager) Stat() {
	ticker := time.NewTicker(config.TcpInterval * time.Second)
	defer ticker.Stop()
	for range ticker.C {
		for _, server := range m.servers {
			m.byteTotalCount += server.ByteCount()
		}
		if m.byteTotalCount == 0 {
			continue
		}
		logger.Logger.Info("[容器] tcp 接收数据统计(KB)", zap.Int("interval", config.TcpInterval), zap.Int64("byteCount", m.byteTotalCount/config.KB))
		m.byteTotalCount = 0
	}
}

type EchoTcpServer struct {
	conn      *net.TCPListener
	byteCount atomic.Int64
}

func (s *EchoTcpServer) Start(port int) bool {
	addr, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", config.IpAddr, port))
	if err != nil {
		logger.Logger.Error("[容器] tcp 服务端口错误", zap.String("ip", config.IpAddr), zap.Int("port", port), zap.Error(err))
		return false
	}
	s.conn, err = net.ListenTCP("tcp", addr)
	if err != nil {
		logger.Logger.Error("[容器] tcp 服务监听错误", zap.String("ip", config.IpAddr), zap.Int("port", port), zap.Error(err))
		return false
	}
	go func() {
		for {
			conn, errAcc := s.conn.Accept()
			if errAcc != nil {
				if errors.Is(errAcc, net.ErrClosed) {
					return
				}
				logger.Logger.Error("[容器] tcp 服务接受连接失败", zap.String("ip", config.IpAddr), zap.Int("port", port), zap.Error(errAcc))
				continue
			}
			go s.handleConnection(conn)
		}
	}()

	return true
}

func (s *EchoTcpServer) Stop() {
	if s.conn != nil {
		_ = s.conn.Close()
	}
}

func (s *EchoTcpServer) handleConnection(conn net.Conn) {
	defer func(conn net.Conn) {
		_ = conn.Close()
	}(conn)
	clientAddr := conn.RemoteAddr().String()
	logger.Logger.Info("[容器] tcp 客户端启动", zap.String("clientAddr", clientAddr))
	const idleTimeout = 3 * time.Minute
	resetIdleTimeout := func() error {
		return conn.SetReadDeadline(time.Now().Add(idleTimeout))
	}
	buf := make([]byte, config.TcpReceiveBuffSize)
	var (
		n   int
		err error
	)
	for {
		if err = resetIdleTimeout(); err != nil {
			logger.Logger.Error("[容器] tcp 重置超时失败", zap.Error(err))
			break
		}
		n, err = conn.Read(buf)
		if err != nil || n <= 0 {
			if errors.Is(err, os.ErrDeadlineExceeded) {
				logger.Logger.Warn("[容器] tcp 3分钟内未收到数据, 关闭连接")
			}
			logger.Logger.Info("[容器] tcp 客户端停止", zap.String("clientAddr", clientAddr))
			break
		}
		s.byteCount.Add(int64(n))
	}
}

func (s *EchoTcpServer) ByteCount() int64 {
	return s.byteCount.And(0)
}
